package com.zc.component.network.tcp.client;

import io.vertx.core.Vertx;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetClientOptions;
import lombok.extern.slf4j.Slf4j;
import com.zc.web.bean.FastBeanCopier;
import com.zc.component.network.core.*;
import com.zc.component.network.core.security.CertificateManager;
import com.zc.component.network.core.security.VertxKeyCertTrustOptions;
import com.zc.component.network.tcp.parser.PayloadParser;
import com.zc.component.network.tcp.parser.PayloadParserBuilder;
import com.zc.core.metadata.ConfigMetadata;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.time.Duration;
import java.util.function.Supplier;

@Component
@Slf4j
public class VertxTcpClientProvider implements NetworkProvider<TcpClientProperties> {

    private final CertificateManager certificateManager;

    private final PayloadParserBuilder payloadParserBuilder;

    private final Vertx vertx;

    public VertxTcpClientProvider(CertificateManager certificateManager, Vertx vertx, PayloadParserBuilder payloadParserBuilder) {
        this.certificateManager = certificateManager;
        this.vertx = vertx;
        this.payloadParserBuilder = payloadParserBuilder;
    }

    @Nonnull
    @Override
    public NetworkType getType() {
        return DefaultNetworkType.TCP_CLIENT;
    }

    @Nonnull
    @Override
    public VertxTcpClientServer createNetwork(@Nonnull TcpClientProperties properties) {
        VertxTcpClientServer clientServe = new VertxTcpClientServer(properties.getId());
        initClient(clientServe, properties);
        return clientServe;
    }

    @Override
    public void reload(@Nonnull Network network, @Nonnull TcpClientProperties properties) {
        initClient(((VertxTcpClientServer) network), properties);
    }

    public void initClient(VertxTcpClientServer clientServer, TcpClientProperties properties) {
        NetClientOptions options = properties.getOptions();
        NetClient netClient = vertx.createNetClient(options);
        clientServer.setClient(netClient);

        Supplier<PayloadParser> supplier = payloadParserBuilder.build(properties.getParserType(), properties);
        supplier.get();
        clientServer.setRecordParser(supplier.get());
        properties.getAddresses().forEach(address -> connect(clientServer, netClient, address, properties, supplier));
    }

    private void connect(
            VertxTcpClientServer clientServer,
            NetClient netClient,
            TcpClientProperties.SocketAddress address,
            TcpClientProperties properties,
            Supplier<PayloadParser> supplier
    ) {
        if (clientServer == null || !clientServer.isAlive()) {
            return;
        }

        try {
            netClient.connect(address.getPort(), address.getHost(), result -> {
                VertxTcpClient client = new VertxTcpClient(properties.getId(),false);
                client.setKeepAliveTimeoutMs(properties.getLong("keepAliveTimeout").orElse(Duration.ofMinutes(10).toMillis()));
                if (result.succeeded()) {
                    log.debug("connect tcp [{}:{}] success", address.getHost(), address.getPort());
                    client.setRecordParser(supplier.get());
                    client.setSocket(result.result());
                    clientServer.sinkConnectClient(client);
                    // 关闭后重连
                    result
                            .result()
                            // 关闭后10秒后重新连接
                            .closeHandler(v -> {
                                client.shutdown();
                                if (!clientServer.isAlive()) {
                                    return;
                                }
                                vertx.setTimer(10000, id ->{
                                    log.error("connection closure, attempt to reconnect [{}:{}]", address.getHost(), address.getPort(),result.cause());
                                    connect(clientServer, netClient, address, properties, supplier);
                                });
                            });

                } else {
                    log.error("connect tcp [{}:{}] error", address.getHost(), address.getPort(),result.cause());
                    client.shutdown();
                    if (!clientServer.isAlive()) {
                        return;
                    }

                    // 10秒后重新连接
                    vertx.setTimer(10000, id -> {
                        log.error("attempt to reconnect [{}:{}]", address.getHost(), address.getPort(),result.cause());
                        connect(clientServer, netClient, address, properties, supplier);
                    } );

                }
            });
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }


    }

    @Nullable
    @Override
    public ConfigMetadata getConfigMetadata() {
        // TODO: 2019/12/19
        return null;
    }

    @Nonnull
    @Override
    public Mono<TcpClientProperties> createConfig(@Nonnull NetworkProperties properties) {
        return Mono.defer(() -> {
            TcpClientProperties config = FastBeanCopier.copy(properties.getConfigurations(), new TcpClientProperties());
            config.setId(properties.getId());
            if (config.getOptions() == null) {
                config.setOptions(new NetClientOptions());
            }
            if (config.isSsl()) {
                config.getOptions().setSsl(true);
                return certificateManager.getCertificate(config.getCertId())
                        .map(VertxKeyCertTrustOptions::new)
                        .doOnNext(config.getOptions()::setKeyCertOptions)
                        .doOnNext(config.getOptions()::setTrustOptions)
                        .thenReturn(config);
            }
            return Mono.just(config);
        });
    }
}
